package org.example.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 从文件中读取数据
 */
object Spark_RDD_File {

  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val context = new SparkContext(conf)

    // 从文件中创建rdd
    // path 路径默认以当前环境的根路径诶基准，可以写绝对路径，也可以写相对路径
    // 也可以是文件的具体路径，也可以写目录
    // 也可以写通配符 datas/1*.txt
    // 也可以写系统路径 hdfs://linux1:8082/test.txt

    val file: RDD[String] = context.textFile("datas/*")
    // textFile也含有分区的概念，他会取最小分区
    // def defaultMinPartitions: Int = math.min(defaultParallelism, 2)
    // Spark的底层读取文件的方式，使用的是Hadoop
    // 在为文件作为数据源读取的时候，hadoop会根据读取的文件的字节数来产生分区，如果你的一个文件的大小为7个字节，

    // totalSize = 7 byte
    // long goalSize = totalSize / (long)(numSplits == 0 ? 1 : numSplits);
    // goalSize = 7 / 2 =3 byte 由于你设置了2个分区，所以是代表一个分区放置3个字节
    // 那么就是 7 / 2 = 3 余1,根据hadoop的规则(1.1)即剩下的大小如果超过了
    // 文件总大小的10%，那么就会产生新的分区，这里1明显大于7的百分之十，所以会产生新的分区，所以是 2+1 三个分区
    // 但是实际上的分区情况，也许会比你设置的大(根据具体的数据分布情况)，但一定不会低于你设置的阈值，所以这是最小分区值
    val file2: RDD[String] = context.textFile("datas/*",2)

    /*
    * 以上说明分区的情况
    * 现在说明数据是如何分区
    * 之前说过了spark读取文件其实是依靠hadoop的文件读取方式，然后我们了解到了，hadoop将会
    * 为我们创建三个分区，每个分区存放3个字节，而我们的文件现在内部的情况是这样的
    * 1.txt @@代表回车换行，只是看不见的字符，但是他确实是占有空间的
    * 文件中的所有的二进制字符对于存放他们的文件而言都是偏移量的，从0开始,
    * 同时偏移量不会被重复读取
    *
    * 一个分区是3个字节，所以第一个是0+3 [0,3]
    *
    * 1@@ [0-3] 012
    * 2@@ [3-6] 345
    * 3   [6-7] 67
    *
    * 那么根据以上规则
    * 0 号分区读取 [0-3] 偏移量，所以会把 1@@2读取，但是因为hadoop的文件读取规则，他会把2的整行读取，所以是1@@2@@
      1 号分区，因为偏移量不会重复读取，所以这个时候已经读到6了，所以他会直接把3读取到
      2 号分区，由于已经没有可读的，所以是空的
    * * */

    file.collect().foreach(println)

    // 关闭

    context.stop()
  }

}
